package com.practice.handler;

import java.util.List;
import lombok.Builder;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.concurrent.TimeUnit;
import cn.hutool.core.util.ObjectUtil;
import java.util.concurrent.Executors;
import cn.hutool.core.collection.CollUtil;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.ScheduledExecutorService;

/**
 * 调度线程处理器
 * 能够批量为指定的元素集合，在指定的时间间隔内，执行指定的方法
 * 调用示例：
 *         // 构造线程调度执行器
 *         ExecutePoolHandler<T, R> executePoolHandler = ExecutePoolHandler.<T, R>builder()
 *                 .all(all)
 *                 .id(id)
 *                 .interval(interval)
 *                 .execute(execute)
 *                 .executeEnum(executeThreadEnum)
 *                 .detectInterval(mcsDeviceProperty.getDetectInterval())
 *                 .executeInterval(mcsDeviceProperty.getExecuteInterval())
 *                 .build();
 *         // 启动执行器
 *         executePoolHandler.start();
 * @param <R> 元素
 * @param <T> 元素id
 * @author: kayn
 * 2025-01-17 21:11
 */
@Slf4j
@Builder
public class ExecutePoolHandler<T,R>{

    /**
     * 调度线程执行方法
     */
    private Consumer<R> execute;

    /**
     * 获取元素id
     */
    private Function<R,T> id;

    /**
     * 获取全部元素
     */
    private Supplier<List<R>> all;

    /**
     * 获取执行间隔
     */
    private Function<R,Integer> interval;

    /**
     * 调度线程类别
     */
    private ExecuteThreadEnum executeEnum;

    /**
     * 默认检测间隔
     */
    @Builder.Default
    private Integer detectInterval = 1000;

    /**
     * 默认执行间隔
     */
    @Builder.Default
    private Integer executeInterval = 500;

    /**
     * 元素列表
     */
    @Builder.Default
    public List<R> elementList = CollUtil.newArrayList();

    /**
     * Executors存储池
     */
    @Builder.Default
    public final ConcurrentHashMapReference<T, ScheduledExecutorService> executorPool = new ConcurrentHashMapReference<>();

    /**
     * 启动处理器核心
     */
    public void start() {

        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

        executor.scheduleAtFixedRate(this::loadThread, 0, detectInterval, TimeUnit.MILLISECONDS);

    }

    /**
     * 加载执行线程
     */
    private void loadThread() {

        try{

            List<R> elementList = all.get();
            List<T> elementIdList = elementList.stream().map(id).toList();

            List<R> newElementList = elementList.stream().filter(r -> !executorPool.containsKey(id.apply(r))).toList();
            newElementList.forEach(this::startThread);
            this.elementList.addAll(newElementList);

            executorPool.keySet().stream().filter(id -> !elementIdList.contains(id)).toList().forEach(this::clearExecutor);
        }catch (Exception e) {
            log.error("【{}线程调度执行器】加载线程异常", executeEnum.getName(),e);
        }
    }

    /**
     * 启动执行线程
     */
    private void startThread(R r) {

        try {

            ScheduledExecutorService executor = executorPool.get(id.apply(r), Executors::newSingleThreadScheduledExecutor);

            Integer executeInterval = ObjectUtil.isNotNull(interval.apply(r)) ? interval.apply(r) : this.executeInterval;

            executor.scheduleWithFixedDelay(() -> execute.accept(r), 0, executeInterval, TimeUnit.MILLISECONDS);

        }catch (Exception e){
            log.error("【{}线程调度执行器】调度线程异常",executeEnum.getName(),e);
        }
    }

    /**
     * 单个清除Executor
     * @param id 线程池id
     */
    public void clearExecutor(T id) {
        try {
            AtomicReference<ScheduledExecutorService> reference = executorPool.get(id);
            if (ObjectUtil.isNotNull(reference)) {
                ScheduledExecutorService executorService = reference.get();
                if (executorService != null) {
                    executorService.shutdownNow();
                }
            }
            executorPool.remove(id);
        }catch (Exception e) {
            log.error("【{}线程调度执行器】清理线程池失败",executeEnum.getName(),e);
        }
    }


    /**
     * 高性能双列集合
     * @param <K>
     * @param <V>
     */
    static class ConcurrentHashMapReference <K,V> extends ConcurrentHashMap<K,AtomicReference<V>> {

        public ConcurrentHashMapReference() {
            super();
        }

        /**
         * 获取V，需要默认值构造方法，池中不存在V时会使用构造方法构造新的V
         * @param key key
         * @param supplier 返回值构造方法
         * @return value
         */
        public V get(K key, Supplier<V> supplier) {

            // 获取容器
            AtomicReference<V> atomicReference = this.computeIfAbsent(key, keyAto -> new AtomicReference<>());

            // 尝试从容器中获取对象
            V value = atomicReference.get();

            // 如果容器中已经有值，直接返回
            if (ObjectUtil.isNotNull(value)) {
                return value;
            }

            // 否则使用compareAndSet方法将默认值设置到容器中(双重保险)
            boolean setResult = atomicReference.compareAndSet(null, supplier.get());

            // 设置成功: 返回新创建的实例(即容器中的值仍然是null)
            if (setResult) {
                return supplier.get();
            }

            // 设置失败: 返回已经存在的实例 (即有其他线程已经设置了值)
            return atomicReference.get();
        }
    }

    @Getter
    enum ExecuteThreadEnum {

        ACTION("action", "动作"),
        ;


        /**
         * 设备类别code
         */
        private final String code;

        /**
         * 设备类别name
         */
        private final String name;

        ExecuteThreadEnum(String code, String name) {
            this.code = code;
            this.name = name;
        }

    }

}
